package xin.marcher.module.inventory.mq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import xin.marcher.module.common.constants.RocketMqConstant;
import xin.marcher.module.inventory.mq.config.RocketMQProperties;
import xin.marcher.module.inventory.mq.consumer.listener.*;

import javax.annotation.Resource;

/**
 * 库存扩容的消息消费
 */
@Configuration
public class ConsumerBeanConfig {

    /**
     * 配置内容对象
     */
    @Resource
    private RocketMQProperties rocketMQProperties;

    /**
     * 库存扩容的消息
     *
     * @param bucketCapacityListener
     * @return
     * @throws MQClientException
     */
    @Bean("bucketCapacityTopic")
    public DefaultMQPushConsumer bucketCapacityTopic(BucketCapacityListener bucketCapacityListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.BUCKET_CAPACITY_GROUP);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.BUCKET_CAPACITY_TOPIC, "*");
        consumer.registerMessageListener(bucketCapacityListener);
        consumer.start();
        return consumer;
    }

    /**
     * 分桶下线的消息
     *
     * @param bucketOfflineListener
     * @return
     * @throws MQClientException
     */
    @Bean("bucketOfflineTopic")
    public DefaultMQPushConsumer bucketOfflineTopic(BucketOfflineListener bucketOfflineListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.BUCKET_OFFLINE_GROUP);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.BUCKET_OFFLINE_TOPIC, "*");
        consumer.registerMessageListener(bucketOfflineListener);
        consumer.start();
        return consumer;
    }

    /**
     * 库存明细记录消息
     *
     * @param inventoryDetailListener
     * @return
     * @throws MQClientException
     */
    @Bean("inventoryDetailTopic")
    public DefaultMQPushConsumer inventoryDetailTopic(InventoryDetailListener inventoryDetailListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.INVENTORY_DETAIL_GROUP);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.INVENTORY_DETAIL_TOPIC, "*");
        consumer.setPullBatchSize(1024);
        consumer.setConsumeMessageBatchMaxSize(1024);
        consumer.registerMessageListener(inventoryDetailListener);
        consumer.start();
        return consumer;
    }

    /**
     * 分桶清空消息
     *
     * @param bucketClearListener
     * @return
     * @throws MQClientException
     */
    @Bean("bucketClearTopic")
    public DefaultMQPushConsumer bucketClearTopic(BucketClearListener bucketClearListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.BUCKET_CLEAR_GROUP);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.BUCKET_CLEAR_TOPIC, "*");
        consumer.registerMessageListener(bucketClearListener);
        consumer.start();
        return consumer;
    }

    /**
     * 刷新本地缓存
     *
     * @param cacheRefreshListener
     * @return
     * @throws MQClientException
     */
    @Bean("cacheRefresTopic")
    public DefaultMQPushConsumer cacheRefresTopic(CacheRefreshListener cacheRefreshListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.REFRESH_CACHE_GROUP);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.DATA_HOT_RADIO_TOPIC, "*");
        consumer.registerMessageListener(cacheRefreshListener);
        consumer.start();
        return consumer;
    }

    /**
     * 同步本地缓存
     *
     * @param bucketRefreshListener
     * @return
     * @throws MQClientException
     */
    @Bean("bucketCacheRefresTopic")
    public DefaultMQPushConsumer BucketCacheRefresTopic(BucketRefreshListener bucketRefreshListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.SYNC_REFRESH_CACHE_GROUP);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.BUCKET_CACHE_REFRESH_TOPIC, "*");
        consumer.registerMessageListener(bucketRefreshListener);
        consumer.start();
        return consumer;
    }

}
